package com.abel.demo.es;

import com.abel.demo.common.BusinessException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.collapse.CollapseBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 索引相当于数据库中的表名
 * mapping相当定义表结构字段类型
 * 文档相当于表中的每行记录数据
 */
@Slf4j
@Component
public class ElasticsearchService {
    @Autowired
    private RestHighLevelClient client;

    private final String DOC_TYPE = "_doc";

    /**
     * 删除索引
     * 注意：删除索引时会把所属的文档数据一并删除
     * @param index
     * @return
     * @throws IOException
     */
    public boolean deleteUserIndex(String index) throws IOException {
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
        AcknowledgedResponse deleteIndexResponse = client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        return deleteIndexResponse.isAcknowledged();
    }

    /**
     * 索引是否存在
     * @param index
     * @return
     * @throws IOException
     */
    public boolean existsIndex(String index) throws IOException {
        GetIndexRequest getRequest = new GetIndexRequest(index);
        boolean exists = client.indices().exists(getRequest, RequestOptions.DEFAULT);
        return exists;
    }

    /**
     * 创建索引
     * @param index 索引名称，相当于数据库表名。必须是小写字母
     * @param builderMapping mapping配置，定义字段的数据类型和分词器指定
     *                       中文IK分词器：[ik_smart：最小分词，ik_max_word：最大分词]
     * @return
     * @throws IOException
     */
    public boolean createIndex(String index, XContentBuilder builderMapping) throws IOException {
        if(existsIndex(index)){
            throw new BusinessException(String.format("索引[%s]已存在", index));
        }

        CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);

        //设置mapping参数
        createIndexRequest.mapping(builderMapping);

        /**
         * 设置分片和副本
         * 分片：当索引上的数据量太大的时候，ES 通过水平拆分的方式将一个索引上的数据拆分出来分配到不同的数据块上，拆分出来的数据库块称之为一个分片
         *      分片相当于mysql数据中的水平分表
         * 副本：副本就是对分片的 Copy，每个主分片都有一个或多个副本分片，当主分片异常时，副本可以提供数据的查询等操作
         */
        createIndexRequest.settings(Settings.builder().put("number_of_shards", 1).put("number_of_replicas", 0));

        CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
        return createIndexResponse.isAcknowledged();
    }

    /**
     * 创建文档
     * @param index 索引
     * @param docId 文档id
     * @param document 文档内容
     * @return
     * @throws IOException
     */
    public RestStatus createDoc(String index, Object document, String docId) throws IOException {
        IndexRequest indexRequest = new IndexRequest(index, DOC_TYPE)
                .id(docId)
                .source(JSON.toJSONString(document), XContentType.JSON);

        IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);

        return indexResponse.status();
    }

    /**
     * 批量生成文档
     * @param index 索引
     * @param documentList 文档集合
     * @param idField 文档对象中的id字段名
     * @return
     * @throws IOException
     */
    public boolean createDoc(String index, List<?> documentList, String idField) throws IOException {
        BulkRequest request = new BulkRequest();

        for (Object doc : documentList) {
            String jsonStr = JSON.toJSONString(doc);
            String id = JSON.parseObject(jsonStr).getString(idField);
            request.add(new IndexRequest(index, DOC_TYPE)
                    .id(id)
                    .source(jsonStr, XContentType.JSON));
        }

        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);

        return !bulkResponse.hasFailures();
    }

    /**
     * 更新文档
     * @param index 索引
     * @param docId 文档id
     * @param document 文档内容
     * @return
     * @throws IOException
     */
    public RestStatus updateDoc(String index, Object document, String docId) throws IOException {
        UpdateRequest updateRequest = new UpdateRequest(index, DOC_TYPE, docId);
        updateRequest.doc(JSON.toJSONString(document), XContentType.JSON);

        UpdateResponse response = client.update(updateRequest, RequestOptions.DEFAULT);

        return response.status();
    }

    /**
     * 删除文档
     * @param index 索引
     * @param docId 文档id
     * @return
     * @throws IOException
     */
    public RestStatus delDoc(String index, String docId) throws IOException {
        if(existsDoc(index, docId)) {
            DeleteRequest deleteRequest = new DeleteRequest(index, DOC_TYPE, docId);
            DeleteResponse response = client.delete(deleteRequest, RequestOptions.DEFAULT);

            return response.status();
        }
        return null;
    }

    /**
     * 文档是否存在
     * @param index
     * @param id
     * @return
     * @throws IOException
     */
    public boolean existsDoc(String index, String id) throws IOException {
        GetRequest request = new GetRequest(index, DOC_TYPE, id);
        GetResponse response = client.get(request, RequestOptions.DEFAULT);

        return response.isExists();
    }

    /**
     * 根据id返回单个文档对象
     * @param index
     * @param id
     * @param clazz
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> T getDoc(String index, String id, Class<T> clazz) throws IOException {
        GetRequest request = new GetRequest(index, DOC_TYPE, id);
        GetResponse response = client.get(request, RequestOptions.DEFAULT);
        if(response.isExists() && !StringUtils.isEmpty(response.getSourceAsString())){
            return JSONObject.parseObject(response.getSourceAsString(), clazz);
        }

        return null;
    }


    /**
     * 无条件，搜索
     * @param index 索引
     * @param size 返回条数
     * @return
     * @throws IOException
     */
    public List<SearchHit> searchAll(String index, int size) throws IOException{
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        SearchRequest searchRequest = new SearchRequest(index);
        searchBuilder.query(QueryBuilders.matchAllQuery()).size(size);
        searchRequest.source(searchBuilder);
        // 同步查询
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();
        List<SearchHit> list = new ArrayList<>();
        for (SearchHit hit: hits){
            list.add(hit);
        }

        return list;
    }

    /**
     * 条件搜索
     * @param index 索引
     * @param queryBuilder 条件构造器
     * @param page 页码
     * @param size 每页返回条数
     * @return
     * @throws IOException
     */
    public SearchHits search(String index, QueryBuilder queryBuilder, int page, int size) throws IOException {
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        SearchRequest searchRequest = new SearchRequest(index);
        searchBuilder.query(queryBuilder)
//                .sort("field", SortOrder.DESC) //排序方式
                .from((page -1) * size)
                .size(size);

        searchRequest.source(searchBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();


        return hits;
    }

    /**
     * 条件搜索并去重，此处返回分页还有点问题，其分页结果是去重前的
     * @param index 索引
     * @param queryBuilder 条件构造器
     * @param distinctField 去重字段
     * @param page
     * @param size
     * @return
     * @throws IOException
     */
    public SearchHits searchAndDistinct(String index, QueryBuilder queryBuilder, String distinctField, int page, int size) throws IOException {
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        SearchRequest searchRequest = new SearchRequest(index);
        searchBuilder.query(queryBuilder)
//                .sort("field", SortOrder.DESC) //排序方式
                .from((page -1) * size)
                .size(size);

        CollapseBuilder collapseBuilder = new CollapseBuilder(distinctField);
        searchBuilder.collapse(collapseBuilder);

//        TermsAggregationBuilder groupBY = AggregationBuilders.terms(groupByField).field(groupByField).size(size);
//        searchBuilder.aggregation(groupBY);

        searchRequest.source(searchBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();

        /*Aggregations aggregations = response.getAggregations();
        Terms terms = aggregations.get(groupByField); //填入上方terms()中相同的分组名
        for (Terms.Bucket bucket : terms.getBuckets()) {
            String key = bucket.getKeyAsString();
        }*/

        return hits;
    }


}
